package ay.shadow.common.rabbitmq.utils;

import ay.shadow.common.rabbitmq.config.RabbitMqClientConfig;
import ay.shadow.core.exception.ShadowException;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.concurrent.TimeoutException;

/**
 * @author shadow
 */
public class RbMqUtil {
    private final ConnectionFactory connectionFactory;

    public RbMqUtil(RabbitMqClientConfig connectionFactory) throws TimeoutException {
        this.connectionFactory = connectionFactory.getMqFactory();
    }

    /**
     * 创建队列
     *
     * @param name       队列名字
     * @param durable    消息是否持久化
     * @param exclusive  是否只自己可见队列（排他队列）
     * @param autoDelete 所有消费者断开连接，是否自动删除队列
     * @param arguments  队列其他构造消息
     */
    public void createQueues(String name, boolean durable, boolean exclusive, boolean autoDelete,
                             Map<String, Object> arguments) {
        try (Connection connection = connectionFactory.newConnection();
             Channel channel = connection.createChannel()
        ) {
            channel.queueDeclare(name, durable, exclusive, autoDelete, arguments);
        } catch (Exception e) {
            throw new ShadowException("创建队列失败");
        }

    }

    /**
     * 删除队列
     *
     * @param queueName 队列名字
     */
    public void deleteQueue(String queueName) {
        try (Connection connection = connectionFactory.newConnection();
             Channel channel = connection.createChannel()) {
            channel.queueDelete(queueName);
        } catch (Exception e) {
            throw new ShadowException("删除队列失败");
        }
    }

    /**
     * 发送消息
     *
     * @param exchange  交换机
     * @param queueName 队列名字
     * @param props     其他参数
     * @param body      消息内容
     */
    public void basicPublish(String exchange, String queueName, AMQP.BasicProperties props, String body) {
        try (Connection connection = connectionFactory.newConnection();
             Channel channel = connection.createChannel()) {
            channel.basicPublish(exchange, queueName, props, body.getBytes(StandardCharsets.UTF_8));
        } catch (Exception e) {
            throw new ShadowException("发送消息失败");
        }

    }

    /**
     * 消费消息
     *
     * @param queueName 队列名字
     * @param autoAck 接收消息是否确认（服务器）
     * @param deliverCallback 一个消息消费回调接口 (实现 DeliverCallback接口)
     * @param cancelCallback 关闭订阅（连接）回调接口
     */
    public void basicConsume(String queueName, boolean autoAck, DeliverCallback deliverCallback, CancelCallback cancelCallback) {
        try {
            Connection connection = connectionFactory.newConnection();
            Channel channel = connection.createChannel();
            channel.basicConsume(queueName, autoAck, deliverCallback, cancelCallback!=null?cancelCallback:consumerTag -> {});
        } catch (Exception e) {
            throw new ShadowException("消费消息失败");
        }

    }

}
